package com.atguigu.edu.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.bean.InteractionCourseCommentBean;
import com.atguigu.edu.realtime.common.bean.TradeCourseOrderBean;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.edu.realtime.common.function.DimAsyncFunction;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import javax.sound.midi.Track;
import java.math.BigDecimal;
import java.util.concurrent.TimeUnit;

public class DwsInteractionCourseComment extends BaseApp {
    public static void main(String[] args) {
        new DwsInteractionCourseComment().start(
                11112,
                4,
                "dws_interaction_course_comment_window",
                "dwd_interaction_course_comment"
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        kafkaStrDS.print();
        // TODO: 2024/10/22 1.流中数据类型转换
        SingleOutputStreamOperator<JSONObject> jsonObj = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String string, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        //  JSONObject jsonObject = JSON.parseObject(string);
                        if (StringUtils.isNotEmpty(string)) {
                            JSONObject jsonObject = JSON.parseObject(string);
                            collector.collect(jsonObject);
                        }
                    }
                }
        );
        // jsonObj.print();
        // TODO: 2024/10/22 2.第二次类型转换   jsonObj -> 封装成统计用的实体bean类
        SingleOutputStreamOperator<InteractionCourseCommentBean> beanDS = jsonObj.map(
                new MapFunction<JSONObject, InteractionCourseCommentBean>() {
                    @Override
                    public InteractionCourseCommentBean map(JSONObject js) throws Exception {
                        String commentId = js.getString("id");
                        String userId = js.getString("user_id");
                        String courseId = js.getString("course_id");
                        Long reviewStars = js.getLong("review_stars");
                        long ts = js.getLong("ts") * 1_000;

                        InteractionCourseCommentBean bean1 = InteractionCourseCommentBean.builder()
                                .courseId(courseId)
                                .reviewTotalStars(reviewStars)
                                .reviewUserCount(1L)
                                .goodReviewUserCount(reviewStars == 5L ? 1L : 0L)
                                .ts(ts)
                                .build();
                        return bean1;
                    }
                }
        );
        //beanDS.print();
        // TODO: 2024/10/22 3. 水位线 + 分组 +开窗 +聚合
        SingleOutputStreamOperator<InteractionCourseCommentBean> watermarks = beanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<InteractionCourseCommentBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                               new SerializableTimestampAssigner<InteractionCourseCommentBean>() {
                                   @Override
                                   public long extractTimestamp(InteractionCourseCommentBean interactionCourseCommentBean, long l) {
                                       return interactionCourseCommentBean.getTs();
                                   }
                               }
                        )
        );
        KeyedStream<InteractionCourseCommentBean, String> keyByBean = watermarks.keyBy(InteractionCourseCommentBean::getCourseId);
        WindowedStream<InteractionCourseCommentBean, String, TimeWindow> windowDS = keyByBean.window(TumblingEventTimeWindows.of(Time.seconds(10)));
        SingleOutputStreamOperator<InteractionCourseCommentBean> reduceDS = windowDS.reduce(
                new ReduceFunction<InteractionCourseCommentBean>() {
                    @Override
                    public InteractionCourseCommentBean reduce(InteractionCourseCommentBean i1, InteractionCourseCommentBean i2) throws Exception {
                        i1.setReviewUserCount(i1.getReviewUserCount() + i2.getReviewUserCount());
                        i1.setGoodReviewUserCount(i1.getGoodReviewUserCount() + i2.getGoodReviewUserCount());
                        i1.setReviewTotalStars(i1.getReviewTotalStars() + i2.getReviewTotalStars());
                        return i1;
                    }
                },
                new ProcessWindowFunction<InteractionCourseCommentBean, InteractionCourseCommentBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<InteractionCourseCommentBean, InteractionCourseCommentBean, String, TimeWindow>.Context context, Iterable<InteractionCourseCommentBean> elements, Collector<InteractionCourseCommentBean> out) throws Exception {
                        String stt = DateFormatUtil.tsToDateTime(context.window().getStart());
                        String edt = DateFormatUtil.tsToDateTime(context.window().getEnd());
                        String curDt = DateFormatUtil.tsToDate(context.window().getStart());
                        for (InteractionCourseCommentBean element : elements) {
                            element.setStt(stt);
                            element.setEdt(edt);
                            element.setCurDate(curDt);
                            element.setTs(System.currentTimeMillis());
                            out.collect(element);
                        }
                    }
                }
        );
        // TODO: 2024/10/22 4. 维度关联     basedb   课程评论行为 + 用户维度
        SingleOutputStreamOperator<InteractionCourseCommentBean> withDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<InteractionCourseCommentBean>() {
                    @Override
                    public void addDims(InteractionCourseCommentBean orderBean, JSONObject dimJsonObj) {
//                        orderBean.setCourseId(dimJsonObj.getString("course_id"));
                        orderBean.setCourseName(dimJsonObj.getString("course_name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_course_info";
                    }

                    @Override
                    public String getRowKey(InteractionCourseCommentBean orderBean) {
                        return orderBean.getCourseId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );

        withDS.print("withDS");

        withDS
                .map(new BeanToJsonStrMapFunction<>())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_interaction_course_comment_window"));

    }
}
